package com.diaohaiwen.framework.mq.rocket.config;

import com.diaohaiwen.framework.mq.rocket.properties.ClientProperties;
import com.diaohaiwen.framework.mq.rocket.properties.ConsumerProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.Set;

import static com.diaohaiwen.framework.mq.rocket.constant.RocketTagConstant.defaultTag;
import static com.diaohaiwen.framework.mq.rocket.constant.RocketTagConstant.tagDelimiter;

/**
 * @author 刁海文
 * @since 2024/1/16 22:08
 */
@Configuration
public class ConsumerConfig extends BaseRocketConfig {

    @Bean
    public DefaultMQPushConsumer consumer(ClientProperties client, GlobalMessageListener listener) throws MQClientException {
        ConsumerProperties properties = client.getConsumer();
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(properties.getConsumerGroup());
        initial(consumer, client);
        consumer.setNamesrvAddr(client.getServerAddr());
        consumer.setMessageModel(properties.getMessageModel());
        consumer.setConsumeFromWhere(properties.getConsumeFromWhere());
        consumer.setConsumeTimeout(properties.getConsumeTimeout());
        consumer.setAllocateMessageQueueStrategy(properties.getAllocateMessageQueueStrategy());
        consumer.setMessageListener(listener);
        consumer.setConsumeThreadMin(Math.max(properties.getConsumeThreadMin(), 20));
        consumer.setConsumeThreadMax(Math.max(20, properties.getConsumeThreadMax()));
        consumer.setAdjustThreadPoolNumsThreshold(properties.getAdjustThreadPoolNumsThreshold());
        consumer.setConsumeConcurrentlyMaxSpan(properties.getConsumeConcurrentlyMaxSpan());
        consumer.setConsumeConcurrentlyMaxSpan(properties.getConsumeConcurrentlyMaxSpan());
        consumer.setPullThresholdForQueue(properties.getPullThresholdForQueue());
        consumer.setPullInterval(properties.getPullInterval());
        consumer.setPullThresholdForTopic(properties.getPullThresholdForTopic());
        consumer.setPullThresholdSizeForTopic(properties.getPullThresholdSizeForTopic());
        consumer.setPullBatchSize(properties.getPullBatchSize());
        consumer.setConsumeMessageBatchMaxSize(properties.getConsumeMessageBatchMaxSize());
        consumer.setPostSubscriptionWhenPull(properties.isPostSubscriptionWhenPull());
        consumer.setUnitMode(properties.isUnitMode());
        consumer.setMaxReconsumeTimes(properties.getMaxReconsumeTimes());
        consumer.setSuspendCurrentQueueTimeMillis(properties.getSuspendCurrentQueueTimeMillis());
        consumer.setConsumeTimeout(properties.getConsumeTimeout());
        consumer.setAwaitTerminationMillisWhenShutdown(properties.getAwaitTerminationMillisWhenShutdown());
        subscribe(consumer);
        consumer.start();
        return consumer;
    }


    public void subscribe(DefaultMQPushConsumer consumer) throws MQClientException {


        for (Map.Entry<String, Map<String, Method>> entry : METHOD_MAP.entrySet()) {

            String topic = entry.getKey();

            Set<String> tags = entry.getValue().keySet();
            String expression = tags.contains(defaultTag) ? defaultTag : String.join(tagDelimiter, tags);
            consumer.subscribe(topic, expression);
        }

    }
}
